package org.myorg.kafka2rabbitmq.demo.rabbitmq;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

public class FlinkRabbitMQConsumer {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		ParameterTool parameterTool = ExecutionEnvUtil.PARAMETER_TOOL;

		final RMQConnectionConfig rmqConnectionConfig = new RMQConnectionConfig
				.Builder()
				.setHost(RabbitMQProducer.host)
				.setPort(RabbitMQProducer.port)
				.setUserName(RabbitMQProducer.username)
				.setPassword(RabbitMQProducer.password)
				.setVirtualHost("/")
				.build();

		DataStreamSource<String> studentDataStreamSource = env.addSource(
				new RMQSource<String>(
						rmqConnectionConfig,
						RabbitMQProducer.QUEUE_NAME,
						true,
						new SimpleStringSchema()))
				.setParallelism(1);
		studentDataStreamSource.print();

		env.execute("consumer from rabbitmq");
	}
}
